package com.atguigu.flink.streamapi.transform;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * Created by Smexy on 2022/11/21
 *
 *      Connect： 连接，联合。
 *                用于把多个不同类型的流，连接到一起后。对不同类型流中的数据，进行不同的处理。
 *                但是处理后的类型得是统一的！
 *
 *
 *       DS<猪肉>              ----> 焯水，打成肉末
 *                  connect                                    ----->  猪肉末
 *       DS<牛肉>              ----> 不焯水，放猪肉精，打成肉末
 *
 *      --------------------------
 *          泾渭分明:
 *                      泾河清
 *                              ---->在黄土高原 汇入 黄河 ---->上游看起来，泾渭分明--->下游都是黄的
 *                      渭河混
 *
 *
 *
 */
public class Demo6_Connect
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //准备多个流
        DataStreamSource<Integer> ds1 = env.fromElements(1, 2, 3);
        DataStreamSource<String> ds4 = env.fromElements("1","2");

        //连接，混合
        ConnectedStreams<Integer, String> ds5 = ds1.connect(ds4);

        //分类处理
        SingleOutputStreamOperator<Double> ds6 = ds5.map(new CoMapFunction<Integer, String, Double>()
        {
            //把 connect 前的流变为输出
            @Override
            public Double map1(Integer value) throws Exception {
                return value + 0d;
            }

            //把 connect 后的流变为输出
            @Override
            public Double map2(String value) throws Exception {
                return Double.valueOf(value);
            }
        });

        //输出
        ds6.print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
